package com.flink.split_stream;

import com.flink.datasource.UserSource;
import com.flink.entity.User;
import com.flink.window.WindowFunctionDemo;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * 描述:
 * 分流
 *
 * @author yanzhengwu
 * @create 2022-08-14 23:27
 */
public class SplitStream {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<User> stream = env.addSource(new UserSource(100))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<User>() {
                            @Override
                            public long extractTimestamp(User element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }));

        //分流就是定义侧输出流的概念和延迟的数据侧输出流是同一种操作
        OutputTag<Tuple3<String, String, Long>> userOne = new OutputTag<Tuple3<String, String, Long>>("用户1") {
        };
        OutputTag<Tuple3<String, String, Long>> userTow = new OutputTag<Tuple3<String, String, Long>>("用户2") {
        };

        //TODO 定义完成outPutTag在数据流中进行分流逻辑
        SingleOutputStreamOperator<User> processStream = stream.process(new ProcessFunction<User, User>() {
            @Override
            public void processElement(User user, Context ctx, Collector<User> out) throws Exception {
                //筛选逻辑给侧输出流也是分成支流
                if (user.getName().equals("用户1")) {
                    ctx.output(userOne, Tuple3.of(user.getName(), user.getProd(), user.getTimestamp()));
                } else if (user.getName().equals("用户2")) {
                    ctx.output(userTow, Tuple3.of(user.getName(), user.getProd(), user.getTimestamp()));
                } else {
                    //不分流的直接给主流输出
                    out.collect(user);
                }
            }
        });

        processStream.print("else->");
        processStream.getSideOutput(userOne).print("userOne->");
        processStream.getSideOutput(userTow).print("userTow->");


        env.execute();


    }
}
